package com.fc.connector.table;

import com.fc.connector.source.MySource;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SocketDynamicTableSource implements ScanTableSource {
  Logger logger = LoggerFactory.getLogger(SocketDynamicTableFactory.class);

  private final String hostname;
  private final int port;
  private final byte byteDelimiter;
  private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
  private final DataType producedDataType;

  public SocketDynamicTableSource(
      String hostname,
      int port,
      byte byteDelimiter,
      DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
      DataType producedDataType) {
    this.hostname = hostname;
    this.port = port;
    this.byteDelimiter = byteDelimiter;
    this.decodingFormat = decodingFormat;
    this.producedDataType = producedDataType;
  }

  @Override
  public ChangelogMode getChangelogMode() {
    // 在我们的例子中，由解码器来决定 changelog 支持的模式
    // 但是在 source 端指定也可以
    return decodingFormat.getChangelogMode();
  }

  @Override
  public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
    logger.info("===============getScanRuntimeProvider===================================");
    // 创建运行时类用于提交给集群

    final DeserializationSchema<RowData> deserializer = decodingFormat.createRuntimeDecoder(
      runtimeProviderContext,
      producedDataType);



    //方法一：使用 SourceFunction 作为数据源
//    final SourceFunction<RowData> sourceFunction = new SocketSourceFunction(
//            hostname,
//            port,
//            byteDelimiter,
//            deserializer);
//
//    return SourceFunctionProvider.of(sourceFunction, false);



//  方法二：使用 Source 作为数据源
    return  SourceProvider.of(new MySource() );
  }

  @Override
  public DynamicTableSource copy() {
    return new SocketDynamicTableSource(hostname, port, byteDelimiter, decodingFormat, producedDataType);
  }

  @Override
  public String asSummaryString() {
    return "Socket Table Source";
  }
}